package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.LogItem;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.zip.*;

@Slf4j
public class LogGZParseImpl {
    private LogGZItemParseImpl  logGZItemParse=new LogGZItemParseImpl();

    public Map<String,List<LogItem>> parse(String folderName) throws IOException {
//        读取文件夹
        Map<String,List<LogItem>> result=new ConcurrentHashMap<>();
        List<Future<LogParseFutureImpl.Model>> futures;
        ExecutorService executorService= Executors.newFixedThreadPool(8);
        try{
            val p= Paths.get(folderName);
            try(val walks= Files.walk(p)){
                if (log.isInfoEnabled()) log.info("{} 文件", walks.count());
                val walk= Files.walk(p);
                futures=walk.parallel().filter(Files::isRegularFile)
                        .filter(path -> path.getFileName().toString().endsWith(".txt"))
                        .map(path -> CompletableFuture.supplyAsync(()->parseFile(path),executorService))
                        .collect(Collectors.toList());
            }
        }catch (IOException e){
            log.error("File does not exists : {}",e.getMessage());
            return result;
        }
        if (log.isInfoEnabled()) log.info("{} futures is running", futures.size());
        all0f(futures);
//        多线程有返回值，parallel是并行执行流
        if (log.isInfoEnabled()) log.info("get result");
        futures.stream().parallel().map(this::getFuture).filter(Objects::nonNull)
                .forEach(x->result.put(x.getPath().toString(),x.getItems()));
        executorService.shutdown();
        return result;
    }
    private LogParseFutureImpl.Model getFuture(Future<LogParseFutureImpl.Model> future) {
        try{
            return future.get();
        }catch(InterruptedException| ExecutionException e){
            log .error(e. getMessage(), e);
            return null;
        }
    }
    //    解析每一个log文件的具体内容
    protected LogParseFutureImpl.Model parseFile(Path path) {
        try(val lines=Files.lines(path)){
            val items=lines.map(logGZItemParse::parse)
                    .filter(Optional::isPresent)
                    .map(Optional::get).collect(Collectors.toList());
            if (log.isDebugEnabled()) log.debug("PARSE :  => {}", path, items.size());
            return new LogParseFutureImpl.Model(path,items);
        }catch (IOException ex){
            log.error(ex.getMessage(), ex);
            return new LogParseFutureImpl.Model(path);
        }
    }
    //    线程阻塞
    private static <T> void all0f(final List<Future<T>> futures){
        val l=futures.size();
        while (true) {
            val b=futures.stream().anyMatch(x->!x.isDone());
            if(!b) break;
            sleep(10);
            if (log.isDebugEnabled()) log.debug("{} futures done", futures.stream().filter(Future::isDone).count());
        }
    }
    private static void sleep(int mills){
        try{
            Thread.sleep(mills);
        }catch(InterruptedException e){
            log.error(e.getMessage());
        }
    }
    @Getter
    protected static class Model{
        public Model(Path path){
            this(path,new ArrayList<>());
        }
        public Model(Path path, List<LogItem> items) {
            this.path = path;this.items = items;
        }
        private final Path path;
        private final List<LogItem> items;
    }
}
